package member

import (
	"encoding/json"
	"fmt"
	"phant-operator/pkg/apis/phant/v1alpha1"
	"phant-operator/pkg/controller"
	"phant-operator/pkg/manager"

	"time"

	"phant-operator/pkg/apis/label"

	"phant-operator/pkg/util"

	"phant-operator/pkg/manager/member/constants"
	mngerutils "phant-operator/pkg/manager/utils"

	apps "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/klog/v2"
	"k8s.io/utils/ptr"
)

type patroniMemberManager struct {
	deps *controller.Dependencies
}

const (
	failoverPath = "/failover"
	configPath   = "/config"
	clusterPath  = "/cluster"
	statusPath   = "/patroni"
	restartPath  = "/restart"

	rolePrimary = "master"
	roleStandby = "replica"

	timeout = 30 * time.Second
)

// NewPatroniManager returns a *patroniManager
func NewPatroniMemberManager(deps *controller.Dependencies) manager.Manager {
	return &patroniMemberManager{
		deps: deps,
	}
}

func (m *patroniMemberManager) Sync(pg *v1alpha1.PostgresCluster) error {
	// Sync Cluster Headless Service
	if err := m.syncClusterHeadlessServiceForPostgresCluster(pg); err != nil {
		return err
	}

	// Sync Cluster Primary Service
	if err := m.syncPostgresServiceForCluster(rolePrimary, pg); err != nil {
		return err
	}

	// Sync Cluster Replica Service
	if err := m.syncPostgresServiceForCluster(roleStandby, pg); err != nil {
		return err
	}

	return m.syncPatroniStatefulSetForPostgresCluster(pg)
}

func (m *patroniMemberManager) syncPatroniStatefulSetForPostgresCluster(pg *v1alpha1.PostgresCluster) error {
	ns := pg.GetNamespace()
	pgName := pg.GetName()

	oldPatroniSetTmp, err := m.deps.StatefulSetLister.StatefulSets(ns).Get(controller.PatroniMemberName(pgName))
	if err != nil && !errors.IsNotFound(err) {
		return fmt.Errorf("syncPDStatefulSetForTidbCluster: fail to get sts %s for cluster %s/%s, error: %s", controller.PatroniMemberName(pgName), ns, pgName, err)
	}
	setNotExist := errors.IsNotFound(err)
	cm, err := m.syncPatroniConfigMap(pg)
	if err != nil {
		return err
	}

	newPatroniSetTmp, err := getNewPatroniSetForPostgresCluster(pg, cm)
	if err != nil {
		return err
	}
	if setNotExist {
		err = mngerutils.SetStatefulSetLastAppliedConfigAnnotation(newPatroniSetTmp)
		if err != nil {
			return err
		}
		if err := m.deps.StatefulSetControl.Create(pg, newPatroniSetTmp); err != nil {
			return err
		}
		//tc.Status.PD.StatefulSet = &apps.StatefulSetStatus{}
		return controller.RequeueErrorf("postgrescluster: [%s/%s], waiting for Patroni running", ns, pgName)
	}
	//to-do: need to conrol sts update proccess with CR status
	if !templateEqual(newPatroniSetTmp, oldPatroniSetTmp) {
		errSTS := mngerutils.UpdateStatefulSet(m.deps.StatefulSetControl, pg, newPatroniSetTmp, oldPatroniSetTmp)
		return controller.RequeueErrorf("postgrescluster: [%s/%s]'s patroni needs to upgrade, %v", ns, pgName, errSTS)
	}
	return nil
}

func getNewPatroniSetForPostgresCluster(pg *v1alpha1.PostgresCluster, cm *corev1.ConfigMap) (*apps.StatefulSet, error) {
	ns := pg.Namespace
	pgName := pg.Name
	basePatroniSpec := pg.BasePartroniSpec()
	instanceName := pg.GetInstanceName()
	annMount, annVolume := annotationsMountVolume()
	dataVolumeName := string(v1alpha1.GetStorageVolumeName("", v1alpha1.PatroniMemberType))
	volMounts := []corev1.VolumeMount{
		annMount,
		{Name: dataVolumeName, MountPath: constants.PatroniDataVolumeMountPath},
	}

	vols := []corev1.Volume{
		annVolume,
	}
	// handle StorageVolumes and AdditionalVolumeMounts in ComponentSpec
	//storageVolMounts, additionalPVCs := util.BuildStorageVolumeAndVolumeMount(tc.Spec.PD.StorageVolumes, tc.Spec.PD.StorageClassName, v1alpha1.PDMemberType)
	//volMounts = append(volMounts, storageVolMounts...)

	storageRequest, err := controller.ParseStorageRequest(pg.Spec.Postgres.Requests)
	if err != nil {
		return nil, fmt.Errorf("cannot parse storage request for PD, tidbcluster %s/%s, error: %v", ns, pgName, err)
	}
	setName := controller.PatroniMemberName(pgName)
	stsLabels := label.New().PatroniCluster(pgName).Instance(instanceName).Patroni()

	podLabels := util.CombineStringMap(stsLabels, pg.GetLabels())
	stsAnnotations := getStsAnnotations(pg.Annotations, label.ComponentLabelKey)
	patroniContainer := corev1.Container{

		Name:            v1alpha1.PatroniMemberType.String(),
		Image:           pg.PatroniImage(),
		ImagePullPolicy: basePatroniSpec.ImagePullPolicy(),
		Command:         []string{"/bin/bash", "/entrypoint.sh"},
		Ports: []corev1.ContainerPort{
			{
				Name: "server",
				//to-do: Support configurable by PostgresCluster CR
				ContainerPort: v1alpha1.DefaultPostgresServerPort,
				Protocol:      corev1.ProtocolTCP,
			},
			{
				Name: "patroni",

				ContainerPort: v1alpha1.DefaultPatroniServerPort,
				Protocol:      corev1.ProtocolTCP,
			},
		},
		VolumeMounts: volMounts,
		Resources:    controller.ContainerResource(pg.Spec.Postgres.ResourceRequirements),
	}
	env := []corev1.EnvVar{
		{
			Name: "PATRONI_KUBERNETES_NAMESPACE",
			ValueFrom: &corev1.EnvVarSource{
				FieldRef: &corev1.ObjectFieldSelector{
					FieldPath: "metadata.namespace",
				},
			},
		},
		{
			Name:  "PATRONI_KUBERNETES_BYPASS_API_SERVICE",
			Value: controller.PatroniMemberName(pgName),
		},
		{
			Name:  "SET_NAME",
			Value: setName,
		},
		{
			Name:  "PATRONI_REPLICATION_USERNAME",
			Value: pg.Spec.Patroni.ReplicationUser.Username,
		},
		{
			Name:  "TZ",
			Value: pg.Spec.Timezone,
		},
	}
	if pg.Spec.Patroni.DCSType == v1alpha1.KubernetesDCSType {
		stsLabelsValue, _ := json.Marshal(stsLabels.PatroniApplication().Labels())
		//env PATRONI_SCOPE and PATRONI_NAME define configure used by Patroni in pods.
		env = append(env,
			[]corev1.EnvVar{
				{
					Name:  "PATRONI_KUBERNETES_BYPASS_API_SERVICE",
					Value: "true",
				},
				{
					Name:  "PATRONI_KUBERNETES_LABELS",
					Value: string(stsLabelsValue),
				},
				{
					Name:  "PATRONI_KUBERNETES_USE_ENDPOINTS",
					Value: "true",
				},
				{
					Name:  "PATRONI_SCOPE",
					Value: pgName,
				},
				{
					Name:  "PATRONI_POSTGRESQL_DATA_DIR",
					Value: constants.PostgresDataVolumeMountPath,
					//Value: "/home/postgres/pgdata",
				},
				{

					Name:  "PATRONI_POSTGRESQL_PGPASS",
					Value: "/tmp/pgpass",
				},
				{
					Name:  "PATRONI_POSTGRESQL_LISTEN",
					Value: "0.0.0.0:5432",
				},
				{
					Name:  "PATRONI_RESTAPI_LISTEN",
					Value: "0.0.0.0:8008",
				},
				{
					Name:  "PATRONI_SUPERUSER_USERNAME",
					Value: "postgres",
				},
				{
					Name: "PATRONI_SUPERUSER_PASSWORD",
					ValueFrom: &corev1.EnvVarSource{
						ConfigMapKeyRef: &corev1.ConfigMapKeySelector{
							LocalObjectReference: corev1.LocalObjectReference{
								Name: cm.Name,
							},
							Key: rolePrimary,
						},
					},
				},
				{
					Name:  "PATRONI_REPLICATION_USERNAME",
					Value: "standby",
				},
				{
					Name: "PATRONI_REPLICATION_PASSWORD",
					ValueFrom: &corev1.EnvVarSource{
						ConfigMapKeyRef: &corev1.ConfigMapKeySelector{
							LocalObjectReference: corev1.LocalObjectReference{
								Name: cm.Name,
							},
							Key: roleStandby,
						},
					},
				},
				{
					Name: "PATRONI_NAME",
					ValueFrom: &corev1.EnvVarSource{
						FieldRef: &corev1.ObjectFieldSelector{
							FieldPath: "metadata.name",
						},
					},
				},
				{
					Name: "PATRONI_KUBERNETES_POD_IP",
					ValueFrom: &corev1.EnvVarSource{
						FieldRef: &corev1.ObjectFieldSelector{
							FieldPath: "status.podIP",
						},
					},
				},
			}...)
	}
	podSpec := basePatroniSpec.BuildPodSpec()
	if basePatroniSpec.HostNetwork() {
		env = append(env, corev1.EnvVar{
			Name: "POD_NAME",
			ValueFrom: &corev1.EnvVarSource{
				FieldRef: &corev1.ObjectFieldSelector{
					FieldPath: "metadata.name",
				},
			},
		})
	}

	if pg.Spec.Postgres.ServiceAccountName != "" {
		podSpec.ServiceAccountName = pg.Spec.Postgres.ServiceAccountName
	}
	patroniContainer.Env = util.AppendEnv(env, basePatroniSpec.Env())
	podSpec.Volumes = vols
	podSpec.Containers, err = MergePatchContainers([]corev1.Container{patroniContainer}, basePatroniSpec.AdditionalContainers())
	if err != nil {
		return nil, fmt.Errorf("failed to merge containers spec for PD of [%s/%s], error: %v", pg.Namespace, pg.Name, err)
	}

	patroniSet := &apps.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:            setName,
			Namespace:       ns,
			Labels:          stsLabels.Labels(),
			Annotations:     stsAnnotations,
			OwnerReferences: []metav1.OwnerReference{controller.GetOwnerRef(pg)},
		},
		Spec: apps.StatefulSetSpec{
			Replicas: ptr.To(pg.Spec.Postgres.Replicas),
			Selector: stsLabels.LabelSelector(),
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: podLabels,
					//Annotations: podAnnotations,
				},
				Spec: podSpec,
			},
			VolumeClaimTemplates: []corev1.PersistentVolumeClaim{
				util.VolumeClaimTemplate(storageRequest, dataVolumeName, pg.Spec.Postgres.StorageClassName),
			},
			ServiceName: controller.ClusterPeerMemberName(pgName),
		},
	}
	return patroniSet, nil
}

// syncPDConfigMap syncs the configmap of PD
func (m *patroniMemberManager) syncPatroniConfigMap(pg *v1alpha1.PostgresCluster) (*corev1.ConfigMap, error) {
	// For backward compatibility, only sync tidb configmap when .pd.config is non-nil
	/*if pg.Spec.Postgres.Config == nil {
		return nil, nil
	}*/
	var (
		cm  *corev1.ConfigMap = getPostgresConfigMap(pg)
		err error
	)
	cm, err = m.deps.Controls.ConfigMapControl.Get(pg, cm)

	if err != nil {
		return nil, err
	}
	return cm, nil
}

func (m *patroniMemberManager) syncClusterHeadlessServiceForPostgresCluster(pg *v1alpha1.PostgresCluster) error {

	if pg.Spec.Paused {
		klog.V(4).Infof("postgres cluster %s/%s is paused, skip syncing for cluster headless service", pg.GetNamespace(), pg.GetName())
		return nil
	}

	ns := pg.GetNamespace()
	pgName := pg.GetName()

	newSvc := getNewPatroniHeadlessServiceForPostgresCluster(pg)
	klog.Infof("sync Patroni HeadlessService %s for PostgresCluster: %s", newSvc.GetName(), pgName)
	oldSvcTmp, err := m.deps.ServiceLister.Services(ns).Get(controller.ClusterPeerMemberName(pgName))
	if errors.IsNotFound(err) {
		err = controller.SetServiceLastAppliedConfigAnnotation(newSvc)
		if err != nil {
			return err
		}
		return m.deps.ServiceControl.Create(pg, newSvc)
	}
	if err != nil {
		return fmt.Errorf("syncPatroniHeadlessServiceForPostgresCluster: failed to get svc %s for cluster %s/%s, error: %s", controller.ClusterPeerMemberName(pgName), ns, pgName, err)
	}

	oldSvc := oldSvcTmp.DeepCopy()

	_, err = m.deps.ServiceControl.SyncComponentService(
		pg,
		newSvc,
		oldSvc,
		false)

	if err != nil {
		return err
	}

	return nil
}

func getNewPatroniHeadlessServiceForPostgresCluster(pg *v1alpha1.PostgresCluster) *corev1.Service {
	ns := pg.Namespace
	pgName := pg.Name
	svcName := controller.ClusterPeerMemberName(pgName)
	instanceName := pg.GetInstanceName()
	patroniSelector := label.New().Instance(instanceName)
	patroniLabels := patroniSelector.Copy().UsedByPeer().Labels()
	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            svcName,
			Namespace:       ns,
			Labels:          patroniLabels,
			OwnerReferences: []metav1.OwnerReference{controller.GetOwnerRef(pg)},
		},
		Spec: corev1.ServiceSpec{
			ClusterIP: "None",
			Ports: []corev1.ServicePort{
				{
					Name:       fmt.Sprintf("tcp-peer-%d", v1alpha1.DefaultPatroniServerPort),
					Port:       v1alpha1.DefaultPatroniServerPort,
					TargetPort: intstr.FromInt(int(v1alpha1.DefaultPatroniServerPort)),
					Protocol:   corev1.ProtocolTCP,
				},
				{
					Name:       fmt.Sprintf("tcp-peer-%d", v1alpha1.DefaultPostgresServerPort),
					Port:       v1alpha1.DefaultPostgresServerPort,
					TargetPort: intstr.FromInt(int(v1alpha1.DefaultPostgresServerPort)),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			Selector:                 patroniSelector,
			PublishNotReadyAddresses: true,
		},
	}
	/*
		if pg.Spec.PreferIPv6 {
			SetServiceWhenPreferIPv6(svc)
		}
	*/

	return svc
}

func (m *patroniMemberManager) syncPostgresServiceForCluster(role string, pg *v1alpha1.PostgresCluster) error {
	if pg.Spec.Paused {
		klog.V(4).Infof("tidb cluster %s/%s is paused, skip syncing for postgres service", pg.GetNamespace(), pg.GetName())
		return nil
	}

	ns := pg.GetNamespace()
	pgName := pg.GetName()

	newSvc := getNewPostgresServiceForCluster(role, pg)
	oldSvcTmp, err := m.deps.ServiceLister.Services(ns).Get(controller.ClusterRoleMemberName(pgName, role))
	if errors.IsNotFound(err) {
		err = controller.SetServiceLastAppliedConfigAnnotation(newSvc)
		if err != nil {
			return err
		}
		return m.deps.ServiceControl.Create(pg, newSvc)
	}
	if err != nil {
		return fmt.Errorf("syncpatroniServiceForTidbCluster: failed to get svc %s for cluster %s/%s, error: %s", newSvc.Name, ns, pgName, err)
	}

	oldSvc := oldSvcTmp.DeepCopy()

	_, err = m.deps.ServiceControl.SyncComponentService(
		pg,
		newSvc,
		oldSvc,
		true)

	if err != nil {
		return err
	}

	return nil
}

func getNewPostgresServiceForCluster(role string, pg *v1alpha1.PostgresCluster) *corev1.Service {
	ns := pg.Namespace
	pgName := pg.Name
	svcName := controller.ClusterRoleMemberName(pgName, role)
	instanceName := pg.GetInstanceName()
	pgSelector := label.New().Instance(instanceName).Role(role)
	pgLabels := pgSelector.Copy().UsedByEndUser().Labels()

	patroniService := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:            svcName,
			Namespace:       ns,
			Labels:          pgLabels,
			OwnerReferences: []metav1.OwnerReference{controller.GetOwnerRef(pg)},
		},
		Spec: corev1.ServiceSpec{
			Type: controller.GetServiceType(pg.Spec.Postgres.Service),
			Ports: []corev1.ServicePort{
				{
					Name:       "server",
					Port:       v1alpha1.DefaultPostgresServerPort,
					TargetPort: intstr.FromInt(int(v1alpha1.DefaultPostgresServerPort)),
					Protocol:   corev1.ProtocolTCP,
				},
			},
			Selector: pgSelector.Labels(),
		},
	}

	// override fields with user-defined ServiceSpec
	svcSpec := pg.Spec.Postgres.Service
	if svcSpec != nil {
		if svcSpec.Type != "" {
			patroniService.Spec.Type = svcSpec.Type
		}
		patroniService.ObjectMeta.Annotations = util.CopyStringMap(svcSpec.Annotations)
		patroniService.ObjectMeta.Labels = util.CombineStringMap(patroniService.ObjectMeta.Labels, svcSpec.Labels)
		if svcSpec.LoadBalancerIP != nil {
			patroniService.Spec.LoadBalancerIP = *svcSpec.LoadBalancerIP
		}
		if svcSpec.ClusterIP != nil {
			patroniService.Spec.ClusterIP = *svcSpec.ClusterIP
		}
		if svcSpec.PortName != nil {
			patroniService.Spec.Ports[0].Name = *svcSpec.PortName
		}
	}
	/*
		if tc.Spec.PreferIPv6 {
			SetServiceWhenPreferIPv6(patroniService)
		}
	*/
	return patroniService
}
